package cn.micro.flink.source;

import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.common.typeinfo.Types;

import org.apache.flink.api.connector.source.util.ratelimit.RateLimiterStrategy;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.connector.datagen.source.DataGeneratorSource;
import org.apache.flink.connector.datagen.source.GeneratorFunction;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;

/**
 * 模拟生成数据---开发环境下可使用
 */
public class GenDataTask {

    public static void main(String[] args) throws Exception {
//        StreamExecutionEnvironment executionEnvironment = StreamExecutionEnvironment.getExecutionEnvironment();
        StreamExecutionEnvironment executionEnvironment = StreamExecutionEnvironment.createLocalEnvironmentWithWebUI(new Configuration());
        executionEnvironment.setParallelism(2);

        //自动生成测试数据
        DataGeneratorSource<String> dataGeneratorSource = new DataGeneratorSource<>(
                (GeneratorFunction<Long, String>) aLong -> aLong+"data",
                Long.MAX_VALUE,
                RateLimiterStrategy.perSecond(10),
                Types.STRING);

        executionEnvironment.fromSource(dataGeneratorSource, WatermarkStrategy.noWatermarks(),"GenDataSource").print();
        //顺序水位线：WatermarkStrategy.forMonotonousTimestamps()
        //处理乱序水位线：WatermarkStrategy.forBoundedOutOfOrderness()


        executionEnvironment.execute();

    }
}
